package db.prob.moviesanddirectors;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Task.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import db.prob.operators.join.MapperJoin;
import db.prob.operators.join.ReducerJoin;
import db.prob.operators.project.MapperProject;
import db.prob.operators.project.ReducerProject;

public class UnsafePlan {
	private static final Log LOG = LogFactory.getLog("UnsafePlan");
	Configuration conf;
	long totalNumOfMessages;
	String inputPath;
	String outputPath;
	int jobNumber;
	
	public UnsafePlan(String inputPath, String outputPath) {
		this.conf = new Configuration();
		this.totalNumOfMessages = 0;
		this.inputPath = inputPath;
		this.outputPath = outputPath;
		this.jobNumber = 1;
	}
	
	private long joinMoviesAndDirectedBy() throws IOException, InterruptedException, ClassNotFoundException {
		conf.set(MapperJoin.TABLE_1, "movies");
		conf.setInt(MapperJoin.TABLE_1_ATTR, 0); // sid
		conf.set(MapperJoin.TABLE_2, "directedby");
		conf.setInt(MapperJoin.TABLE_2_ATTR, 1); // sid
		conf.set(MapperJoin.RESULT_TABLE, "R");
		
		Job job = new Job(conf);
		job.setJobName("Join movies and directedby");

		job.setMapperClass(MapperJoin.class);
        job.setReducerClass(ReducerJoin.class);
		
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

		Path inputPath  = new Path(this.inputPath);
		Path outputPath = new Path(this.inputPath + jobNumber);
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		job.waitForCompletion(true);
		this.jobNumber++;
		
		return job.getCounters().findCounter(Counter.MAP_INPUT_RECORDS).getValue();

	}
	
	
	private long joinRAndDirectors() throws IOException, InterruptedException, ClassNotFoundException {
		conf.set(MapperJoin.TABLE_1, "R");
		conf.setInt(MapperJoin.TABLE_1_ATTR, 1); // cid
		conf.set(MapperJoin.TABLE_2, "directors");
		conf.setInt(MapperJoin.TABLE_2_ATTR, 0);
		conf.set(MapperJoin.RESULT_TABLE, "RR");
		
		Job job = new Job(conf);
		job.setJobName("Join R and directors");

		job.setMapperClass(MapperJoin.class);
        job.setReducerClass(ReducerJoin.class);
		
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

		Path inputPath  = new Path(this.inputPath + (jobNumber - 1));
		Path outputPath = new Path(this.inputPath + jobNumber);
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		job.waitForCompletion(true);
		this.jobNumber++;
		
		return job.getCounters().findCounter(Counter.MAP_INPUT_RECORDS).getValue();
	}
	
	private long projectRR() throws IOException, InterruptedException, ClassNotFoundException {
		// projection on 'students' relation.
		conf.set(MapperProject.TABLE_NAME,"RR");
		conf.set(MapperProject.COLUMNS,"1");
		conf.set(MapperProject.RESULT_TABLE,"result");
		
		Job job = new Job(conf);
		job.setJobName("Project R");
		
		job.setMapperClass(MapperProject.class);
        job.setReducerClass(ReducerProject.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

		Path inputPath  = new Path(this.inputPath + (jobNumber - 1));
		Path outputPath = new Path(this.inputPath + jobNumber);
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		job.waitForCompletion(true);
		this.jobNumber++;
		
		return job.getCounters().findCounter(Counter.MAP_INPUT_RECORDS).getValue();

	}
	
	
	/**
	 * An unsafe execution plan.
	 * 
	 * 1. Join students and sic = R'
	 * 2. Project R' on cid, gender = R''
	 * 3. Join R'' and courses
	 * 
	 * @param args
	 * @throws IOException 
	 * @throws ClassNotFoundException 
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		long numOfJobMessages = 0;
		
		UnsafePlan plan = new UnsafePlan(args[0],args[1]);
		
		numOfJobMessages = plan.joinMoviesAndDirectedBy();
		System.out.format("Number of messages: %d\n", numOfJobMessages);
		plan.totalNumOfMessages += numOfJobMessages;

		numOfJobMessages = plan.joinRAndDirectors();
		System.out.format("Number of messages: %d\n", numOfJobMessages);
		plan.totalNumOfMessages += numOfJobMessages;
				
		numOfJobMessages = plan.projectRR();
		System.out.format("Number of messages: %d\n", numOfJobMessages);
		plan.totalNumOfMessages += numOfJobMessages;
		
		System.out.format("Total number of messages: %d\n", plan.totalNumOfMessages);
	}

}
